#define _POSIX_C_SOURCE 200112L
#include "../../include/gudb.h"
#include "../storage/file_system.h"
#include "../parser/sql_parser.h"
#include "../planner/query_planner.h"
#include "../executor/executor.h"
#include "../executor/trigger_system.h"
#include "../storage/storage_engine.h"
#include "../memory/memory_manager.h"
#include "../wal/wal_manager.h"
#include "../network/network_protocol.h"
#include "../metadata/metadata_manager.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <errno.h>
#include <time.h>
#include <pthread.h>
#include <inttypes.h>

// Global error messages (for future use)
// static const char* error_messages[] = {
//     "Success",
//     "Syntax error",
//     "Out of memory",
//     "I/O error",
//     "Transaction conflict",
//     "Authentication failed",
//     "Permission denied"
// };

// Thread-local error storage
static __thread char last_error[256] = {0};

// Internal helper functions
static int init_page_cache(gudb* db);
static int init_wal_system(gudb* db);
static int load_metadata(gudb* db);
static int authenticate_user(gudb* db, const char* username, const char* password);
static void set_error(const char* msg);
// static uint32_t crc32(const void* data, size_t len);
static uint64_t get_timestamp_ms(void);
static tuple_t* create_tuple_from_json(const char* json_data, column_info_t* columns, uint32_t column_count);

// Database open function
gudb* gudb_open(const char* path, int flags, const char* user, const char* password) {
    if (!path || !user || !password) {
        set_error("Invalid parameters");
        return NULL;
    }

    // Create database structure
    gudb* db = calloc(1, sizeof(gudb));
    if (!db) {
        set_error("Out of memory");
        return NULL;
    }

    // Initialize basic fields
    strncpy(db->data_dir, path, sizeof(db->data_dir) - 1);
    db->port = GUDB_DEFAULT_PORT;
    db->max_connections = GUDB_MAX_CONNECTIONS;
    db->next_xid = 1;
    db->uptime_seconds = time(NULL);

    // Create data directory if needed
    if (flags & GU_OPEN_CREATE) {
        if (mkdir(path, 0755) != 0 && errno != EEXIST) {
            set_error("Failed to create data directory");
            free(db);
            return NULL;
        }
    }

    // Initialize file system
    db->fs = fs_init(path);
    if (!db->fs) {
        set_error("Failed to initialize file system");
        free(db);
        return NULL;
    }

    // Set legacy file descriptors for compatibility
    db->data_fd = fs_get_data_fd(db->fs);
    db->wal_fd = fs_get_wal_fd(db->fs);
    db->binlog_fd = fs_get_binlog_fd(db->fs);
    db->meta_fd = fs_get_meta_fd(db->fs);
    db->log_fd = fs_get_log_fd(db->fs);
    db->lock_fd = fs_get_lock_fd(db->fs);

    // Initialize memory manager
    db->memory_mgr = mm_init(GUDB_PAGE_CACHE_SIZE);
    if (!db->memory_mgr) {
        gudb_close(db);
        return NULL;
    }
    
    // Initialize storage engine
    db->storage = storage_init((memory_manager_t*)db->memory_mgr, db->fs);
    if (!db->storage) {
        gudb_close(db);
        return NULL;
    }
    
    // Initialize metadata manager
    db->metadata = metadata_manager_create(path);
    if (!db->metadata) {
        gudb_close(db);
        return NULL;
    }
    
    if (metadata_manager_init(db->metadata) != 0) {
        gudb_close(db);
        return NULL;
    }
    
    // Initialize WAL manager
    char wal_dir[512];
    snprintf(wal_dir, sizeof(wal_dir), "%s/wal", path);
    db->wal_manager = wal_manager_create(wal_dir);
    if (!db->wal_manager) {
        gudb_close(db);
        return NULL;
    }
    
    // Note: WAL manager background threads not started for basic operation
    // Full WAL functionality including background threads can be enabled separately
    // if (wal_manager_start((wal_manager_t*)db->wal_manager) != 0) {
    //     gudb_close(db);
    //     return NULL;
    // }

    // Initialize memory subsystems (legacy)
    if (init_page_cache(db) != GU_OK) {
        gudb_close(db);
        return NULL;
    }

    if (init_wal_system(db) != GU_OK) {
        gudb_close(db);
        return NULL;
    }

    // Load metadata and authenticate user
    if (load_metadata(db) != GU_OK) {
        gudb_close(db);
        return NULL;
    }

    if (authenticate_user(db, user, password) != GU_OK) {
        gudb_close(db);
        return NULL;
    }

    // Log successful database open
    char log_msg[256];
    snprintf(log_msg, sizeof(log_msg), "Database opened successfully by user: %s", user);
    ssize_t written = write(db->log_fd, log_msg, strlen(log_msg));
    (void)written; // Suppress unused result warning
    written = write(db->log_fd, "\n", 1);
    (void)written; // Suppress unused result warning

    return db;
}

// Database close function
int gudb_close(gudb* db) {
    if (!db) {
        return GU_ERR_SYNTAX;
    }

    // Stop network server
    if (db->network_server) {
        gudb_server_stop(db);
    }
    
    // Clean up WAL manager
    if (db->wal_manager) {
        wal_manager_destroy((wal_manager_t*)db->wal_manager);
        db->wal_manager = NULL;
    }

    // Clean up storage engine
    if (db->storage) {
        storage_cleanup((bw_tree_t*)db->storage);
        db->storage = NULL;
    }
    
    // Clean up metadata manager
    if (db->metadata) {
        metadata_manager_destroy(db->metadata);
        db->metadata = NULL;
    }

    // Clean up memory manager
    if (db->memory_mgr) {
        mm_cleanup((memory_manager_t*)db->memory_mgr);
        db->memory_mgr = NULL;
    }

    // Clean up file system
    if (db->fs) {
        fs_cleanup(db->fs);
        db->fs = NULL;
    }

    // Free legacy memory
    if (db->page_cache) {
        free(db->page_cache);
        db->page_cache = NULL;
    }
    if (db->wal_buffer) {
        free(db->wal_buffer);
        db->wal_buffer = NULL;
    }
    if (db->binlog_buffer) {
        free(db->binlog_buffer);
        db->binlog_buffer = NULL;
    }

    free(db);
    return GU_OK;
}

// Transaction begin
gudb_txn* gudb_begin(gudb* db, int isolation_level, const char* database_name) {
    if (!db) {
        set_error("Invalid database handle");
        return NULL;
    }

    gudb_txn* txn = calloc(1, sizeof(gudb_txn));
    if (!txn) {
        set_error("Out of memory");
        return NULL;
    }

    txn->xid = __sync_fetch_and_add(&db->next_xid, 1);
    txn->db = db;
    txn->isolation_level = isolation_level;
    txn->start_time = get_timestamp_ms();
    
    // Set current database (default to "gudb" if not specified)
    if (database_name && strlen(database_name) > 0) {
        strncpy(txn->current_database, database_name, sizeof(txn->current_database) - 1);
        txn->current_database[sizeof(txn->current_database) - 1] = '\0';
    } else {
        strncpy(txn->current_database, "gudb", sizeof(txn->current_database) - 1);
    }
    
    // Initially no database is explicitly selected by user (only USE command sets this to true)
    txn->database_explicitly_selected = false;
    
    // Log transaction begin to WAL
    if (db->wal_manager) {
        isolation_level_t wal_isolation;
        switch (isolation_level) {
            case GU_TXN_RC:  wal_isolation = ISOLATION_READ_COMMITTED; break;
            case GU_TXN_RR:  wal_isolation = ISOLATION_REPEATABLE_READ; break;
            case GU_TXN_SSI: wal_isolation = ISOLATION_SERIALIZABLE; break;
            default:         wal_isolation = ISOLATION_READ_COMMITTED; break;
        }
        wal_log_transaction_begin((wal_manager_t*)db->wal_manager, txn->xid, wal_isolation);
    }

    // Find empty slot for active transaction
    for (int i = 0; i < GUDB_MAX_ACTIVE_TXN; i++) {
        if (!db->active_txns[i]) {
            db->active_txns[i] = txn;
            break;
        }
    }

    return txn;
}

// Transaction commit
int gudb_commit(gudb_txn* txn) {
    if (!txn || !txn->db) {
        return GU_ERR_SYNTAX;
    }

    // Log transaction commit to WAL
    if (txn->db->wal_manager) {
        lsn_t commit_lsn = wal_log_transaction_commit((wal_manager_t*)txn->db->wal_manager, txn->xid);
        // Note: WAL sync disabled for basic operation when background threads are not running
        // In production, WAL sync would ensure durability
        (void)commit_lsn; // Suppress unused variable warning
    }

    // Remove from active transactions
    for (int i = 0; i < GUDB_MAX_ACTIVE_TXN; i++) {
        if (txn->db->active_txns[i] == txn) {
            txn->db->active_txns[i] = NULL;
            break;
        }
    }

    free(txn);
    return GU_OK;
}

// Transaction rollback
int gudb_rollback(gudb_txn* txn) {
    if (!txn || !txn->db) {
        return GU_ERR_SYNTAX;
    }

    // Log transaction abort to WAL
    if (txn->db->wal_manager) {
        wal_log_transaction_abort((wal_manager_t*)txn->db->wal_manager, txn->xid);
    }

    // Remove from active transactions
    for (int i = 0; i < GUDB_MAX_ACTIVE_TXN; i++) {
        if (txn->db->active_txns[i] == txn) {
            txn->db->active_txns[i] = NULL;
            break;
        }
    }

    free(txn);
    return GU_OK;
}

// Helper function to create sample data for common tables
static result_t* create_sample_data_for_table(const char* table_name) {
    result_t* res = calloc(1, sizeof(result_t));
    if (!res) return NULL;
    
    if (strcasecmp(table_name, "products") == 0) {
        // Create products table data
        res->col_count = 4;
        res->column_names = calloc(4, sizeof(char*));
        res->column_names[0] = strdup("id");
        res->column_names[1] = strdup("name");
        res->column_names[2] = strdup("price");
        res->column_names[3] = strdup("category");
        
        // Add sample rows
        res->row_count = 3;
        res->rows = calloc(3, sizeof(char**));
        
        // Row 1
        res->rows[0] = calloc(4, sizeof(char*));
        res->rows[0][0] = strdup("1");
        res->rows[0][1] = strdup("Laptop");
        res->rows[0][2] = strdup("999.99");
        res->rows[0][3] = strdup("Electronics");
        
        // Row 2
        res->rows[1] = calloc(4, sizeof(char*));
        res->rows[1][0] = strdup("2");
        res->rows[1][1] = strdup("Phone");
        res->rows[1][2] = strdup("699.99");
        res->rows[1][3] = strdup("Electronics");
        
        // Row 3
        res->rows[2] = calloc(4, sizeof(char*));
        res->rows[2][0] = strdup("3");
        res->rows[2][1] = strdup("Book");
        res->rows[2][2] = strdup("19.99");
        res->rows[2][3] = strdup("Books");
        
    } else if (strcasecmp(table_name, "users") == 0) {
        // Create users table data
        res->col_count = 3;
        res->column_names = calloc(3, sizeof(char*));
        res->column_names[0] = strdup("id");
        res->column_names[1] = strdup("username");
        res->column_names[2] = strdup("email");
        
        // Add sample rows
        res->row_count = 2;
        res->rows = calloc(2, sizeof(char**));
        
        // Row 1
        res->rows[0] = calloc(3, sizeof(char*));
        res->rows[0][0] = strdup("1");
        res->rows[0][1] = strdup("admin");
        res->rows[0][2] = strdup("admin@gudb.com");
        
        // Row 2
        res->rows[1] = calloc(3, sizeof(char*));
        res->rows[1][0] = strdup("2");
        res->rows[1][1] = strdup("guest");
        res->rows[1][2] = strdup("guest@gudb.com");
        
    } else {
        // Default empty table
        res->col_count = 1;
        res->column_names = calloc(1, sizeof(char*));
        res->column_names[0] = strdup("message");
        res->row_count = 1;
        res->rows = calloc(1, sizeof(char**));
        res->rows[0] = calloc(1, sizeof(char*));
        res->rows[0][0] = strdup("Table not found or empty");
    }
    
    return res;
}

// Helper function to handle SHOW commands using metadata manager
static result_t* handle_show_command_with_context(gudb_txn* txn, const char* sql) {
    result_t* res = calloc(1, sizeof(result_t));
    if (!res) return NULL;
    
    if (strncasecmp(sql, "show databases", 14) == 0) {
        res->col_count = 1;
        res->column_names = calloc(1, sizeof(char*));
        res->column_names[0] = strdup("Database");
        
        if (txn->db->metadata) {
            database_info_t* databases = NULL;
            uint32_t db_count = 0;
            
            if (metadata_list_databases(txn->db->metadata, &databases, &db_count) == 0 && db_count > 0) {
                res->row_count = db_count;
                res->rows = calloc(db_count, sizeof(char**));
                
                for (uint32_t i = 0; i < db_count; i++) {
                    res->rows[i] = calloc(1, sizeof(char*));
                    res->rows[i][0] = strdup(databases[i].name);
                }
                
                free(databases);
            } else {
                // Fallback: no databases found
                res->row_count = 1;
                res->rows = calloc(1, sizeof(char**));
                res->rows[0] = calloc(1, sizeof(char*));
                res->rows[0][0] = strdup("No databases found");
            }
        } else {
            // Fallback: metadata manager not available
            res->row_count = 1;
            res->rows = calloc(1, sizeof(char**));
            res->rows[0] = calloc(1, sizeof(char*));
            res->rows[0][0] = strdup("Metadata unavailable");
        }
        
    } else if (strncasecmp(sql, "show tables", 11) == 0) {
        // Check if user has explicitly selected a database
        if (!txn->database_explicitly_selected) {
            res->error_msg = strdup("No database selected. Use 'USE database_name' first.");
            return res;
        }
        
        res->col_count = 1;
        res->column_names = calloc(1, sizeof(char*));
        
        if (txn->db->metadata) {
            table_info_t* tables = NULL;
            uint32_t table_count = 0;
            
            // printf("DEBUG: SHOW TABLES - calling metadata_list_tables\\n");
            // Use the current database context
            const char* current_db = txn->current_database;
            
            // Dynamic column name based on current database
            char column_name[128];
            snprintf(column_name, sizeof(column_name), "Tables_in_%s", current_db);
            res->column_names[0] = strdup(column_name);
            
            int list_result = metadata_list_tables(txn->db->metadata, current_db, &tables, &table_count);
            // printf("DEBUG: metadata_list_tables returned %d, table_count=%u\\n", list_result, table_count);
            
            if (list_result == 0 && table_count > 0) {
                // printf("DEBUG: Found %u tables\\n", table_count);
                res->row_count = table_count;
                res->rows = calloc(table_count, sizeof(char**));
                
                for (uint32_t i = 0; i < table_count; i++) {
                    res->rows[i] = calloc(1, sizeof(char*));
                    res->rows[i][0] = strdup(tables[i].table_name);
                    // printf("DEBUG: Added table: %s\\n", tables[i].table_name);
                }
                
                free(tables);
            } else {
                // Fallback: no tables found
                // printf("DEBUG: No tables found, showing fallback message\\n");
                res->row_count = 0;
                res->rows = NULL;
            }
        } else {
            // Fallback: metadata manager not available
            // printf("DEBUG: Metadata manager not available\\n");
            const char* current_db = txn->current_database;
            
            // Dynamic column name based on current database
            char column_name[128];
            snprintf(column_name, sizeof(column_name), "Tables_in_%s", current_db);
            res->column_names[0] = strdup(column_name);
            
            res->row_count = 1;
            res->rows = calloc(1, sizeof(char**));
            res->rows[0] = calloc(1, sizeof(char*));
            res->rows[0][0] = strdup("Metadata unavailable");
        }
        
    } else {
        res->col_count = 1;
        res->column_names = calloc(1, sizeof(char*));
        res->column_names[0] = strdup("result");
        res->row_count = 1;
        res->rows = calloc(1, sizeof(char**));
        res->rows[0] = calloc(1, sizeof(char*));
        res->rows[0][0] = strdup("SHOW command not implemented");
    }
    
    return res;
}

// SQL execution (simplified implementation)
int gudb_exec(gudb_txn* txn, const char* sql, const char** params __attribute__((unused)), result_t** result) {
    if (!txn || !sql || !result) {
        return GU_ERR_SYNTAX;
    }

    // First check for SHOW commands (handle before parsing since they might not be in the parser)
    if (strncasecmp(sql, "show ", 5) == 0) {
        *result = handle_show_command_with_context(txn, sql);
        return *result ? GU_OK : GU_ERR_NOMEM;
    }
    
    // Step 1: Parse SQL
    parser_t* parser = parser_create(sql);
    if (!parser) {
        return GU_ERR_NOMEM;
    }
    
    statement_t* stmt = parser_parse_statement(parser);
    if (parser_has_error(parser) || !stmt) {
        parser_destroy(parser);
        return GU_ERR_SYNTAX;
    }
    
    // Create result structure
    result_t* res = calloc(1, sizeof(result_t));
    if (!res) {
        statement_destroy(stmt);
        parser_destroy(parser);
        return GU_ERR_NOMEM;
    }
    
    // Execute based on statement type
    int exec_result = GU_OK;
    switch (stmt->type) {
        case STMT_SELECT: {
            // Real SELECT implementation reading from storage
            if (stmt->data.select.from_list && stmt->data.select.from_list->table_name) {
                char* table_name = stmt->data.select.from_list->table_name;
                
                // Check if table exists
                if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                    exec_result = GU_ERR_SYNTAX; // Table not found
                    break;
                }
                
                // Get table column information
                column_info_t* columns = NULL;
                uint32_t column_count = 0;
                if (metadata_get_table_columns(txn->db->metadata, txn->current_database, table_name, &columns, &column_count) != 0 || !columns) {
                    exec_result = GU_ERR_IO;
                    break;
                }
                
                // Set up result structure with table columns
                res->col_count = column_count;
                res->column_names = calloc(column_count, sizeof(char*));
                for (uint32_t i = 0; i < column_count; i++) {
                    res->column_names[i] = strdup(columns[i].column_name);
                }
                
                // Scan storage for all records of this table
                // Keys are stored as "table_name:row_id"
                char key_prefix[256];
                snprintf(key_prefix, sizeof(key_prefix), "%s:", table_name);
                
                // Simple implementation: try to find rows by scanning row IDs
                // This is not efficient but works for demonstration
                uint32_t max_rows_to_check = 1000; // Reasonable limit
                uint32_t found_rows = 0;
                char*** temp_rows = calloc(max_rows_to_check, sizeof(char**));
                
                for (uint32_t row_id = 1; row_id <= max_rows_to_check && found_rows < max_rows_to_check; row_id++) {
                    char search_key[256];
                    snprintf(search_key, sizeof(search_key), "%s:%u", table_name, row_id);
                    
                    search_result_t search_result = storage_search((bw_tree_t*)txn->db->storage, search_key, strlen(search_key));
                    if (search_result.found && search_result.value) {
                        // Parse JSON-like value: {"col1":"val1","col2":"val2"}
                        temp_rows[found_rows] = calloc(column_count, sizeof(char*));
                        
                        // Simple JSON parsing for each column
                        for (uint32_t col = 0; col < column_count; col++) {
                            char search_pattern[256];
                            snprintf(search_pattern, sizeof(search_pattern), "\"%s\":\"", columns[col].column_name);
                            
                            char* start = strstr(search_result.value, search_pattern);
                            if (start) {
                                start += strlen(search_pattern);
                                char* end = strchr(start, '"');
                                if (end) {
                                    size_t value_len = end - start;
                                    temp_rows[found_rows][col] = malloc(value_len + 1);
                                    strncpy(temp_rows[found_rows][col], start, value_len);
                                    temp_rows[found_rows][col][value_len] = '\0';
                                } else {
                                    temp_rows[found_rows][col] = strdup("NULL");
                                }
                            } else {
                                temp_rows[found_rows][col] = strdup("NULL");
                            }
                        }
                        found_rows++;
                    }
                }
                
                // Copy found rows to result structure
                res->row_count = found_rows;
                if (found_rows > 0) {
                    res->rows = calloc(found_rows, sizeof(char**));
                    for (uint32_t i = 0; i < found_rows; i++) {
                        res->rows[i] = temp_rows[i];
                    }
                }
                
                free(temp_rows);
                free(columns);
                
            } else {
                // Handle simple SELECT queries without FROM clause (like SELECT 1)
                res->col_count = 1;
                res->column_names = calloc(1, sizeof(char*));
                res->column_names[0] = strdup("result");
                res->row_count = 1;
                res->rows = calloc(1, sizeof(char**));
                res->rows[0] = calloc(1, sizeof(char*));
                res->rows[0][0] = strdup("1");
            }
            break;
        }
        
        case STMT_INSERT: {
            // Real INSERT implementation with metadata validation
            if (stmt->data.insert.table_name && txn->db->metadata && txn->db->storage) {
                char* table_name = stmt->data.insert.table_name;
                
                // Check if table exists
                if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                    exec_result = GU_ERR_SYNTAX; // Table not found
                    break;
                }
                
                // Get table column information
                column_info_t* columns = NULL;
                uint32_t column_count = 0;
                if (metadata_get_table_columns(txn->db->metadata, txn->current_database, table_name, &columns, &column_count) != 0 || !columns) {
                    exec_result = GU_ERR_IO;
                    break;
                }
                
                // Validate value count matches column count
                if (stmt->data.insert.value_count != column_count) {
                    free(columns);
                    exec_result = GU_ERR_SYNTAX;
                    break;
                }
                
                // Check primary key constraints - REAL implementation
                // Find primary key columns and check for duplicates in storage
                for (uint32_t i = 0; i < column_count; i++) {
                    if (columns[i].is_primary_key && i < stmt->data.insert.value_count) {
                        expression_t* pk_value_expr = stmt->data.insert.values[i];
                        if (pk_value_expr && pk_value_expr->type == EXPR_LITERAL && 
                            pk_value_expr->data.literal.value) {
                            
                            char* pk_value = pk_value_expr->data.literal.value;
                            
                            // Scan existing data for same primary key value
                            uint32_t max_rows_to_check = 1000;
                            for (uint32_t row_id = 1; row_id <= max_rows_to_check; row_id++) {
                                char search_key[256];
                                snprintf(search_key, sizeof(search_key), "%s:%u", table_name, row_id);
                                
                                search_result_t search_result = storage_search((bw_tree_t*)txn->db->storage, 
                                                                            search_key, strlen(search_key));
                                if (search_result.found && search_result.value) {
                                    // Parse JSON to extract primary key value
                                    char search_pattern[256];
                                    snprintf(search_pattern, sizeof(search_pattern), "\"%s\":\"", columns[i].column_name);
                                    
                                    char* start = strstr(search_result.value, search_pattern);
                                    if (start) {
                                        start += strlen(search_pattern);
                                        char* end = strchr(start, '"');
                                        if (end) {
                                            size_t existing_value_len = end - start;
                                            if (strlen(pk_value) == existing_value_len && 
                                                strncmp(pk_value, start, existing_value_len) == 0) {
                                                // Primary key violation found!
                                                free(columns);
                                                exec_result = GU_ERR_SYNTAX; // Primary key constraint violation
                                                goto insert_cleanup;
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
                
                // Get next row ID
                int row_id = metadata_get_next_row_id(txn->db->metadata, txn->current_database, table_name);
                if (row_id <= 0) {
                    free(columns);
                    exec_result = GU_ERR_IO;
                    break;
                }
                
                // Build row data in JSON-like format
                char value_buf[2048];
                int offset = snprintf(value_buf, sizeof(value_buf), "{");
                
                for (uint32_t i = 0; i < stmt->data.insert.value_count && i < column_count; i++) {
                    expression_t* value_expr = stmt->data.insert.values[i];
                    
                    if (value_expr && value_expr->type == EXPR_LITERAL) {
                        // Add comma separator (except for first item)
                        if (i > 0) {
                            offset += snprintf(value_buf + offset, sizeof(value_buf) - offset, ",");
                        }
                        
                        // Add column name and value
                        offset += snprintf(value_buf + offset, sizeof(value_buf) - offset, 
                                         "\"%s\":\"%s\"", 
                                         columns[i].column_name, 
                                         value_expr->data.literal.value ? value_expr->data.literal.value : "NULL");
                    }
                }
                offset += snprintf(value_buf + offset, sizeof(value_buf) - offset, "}");
                
                // Create storage key: table_name:row_id
                char key_buf[256];
                snprintf(key_buf, sizeof(key_buf), "%s:%d", table_name, row_id);
                
                // Log to WAL before making the change
                if (txn->db->wal_manager) {
                    uint32_t table_id = 1;
                    uint32_t page_id = 0;
                    uint64_t tuple_id = (uint64_t)row_id;
                    
                    wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                 txn->xid, table_id, page_id, tuple_id,
                                 value_buf, strlen(value_buf));
                }
                
                // Insert into storage
                int insert_result = storage_insert((bw_tree_t*)txn->db->storage, 
                                                 key_buf, strlen(key_buf),
                                                 value_buf, strlen(value_buf));
                if (insert_result != 0) {
                    free(columns);
                    exec_result = GU_ERR_IO;
                    break;
                }
                
                // Update row count in metadata
                if (metadata_update_row_count(txn->db->metadata, txn->current_database, table_name, 
                                             metadata_get_table_row_count(txn->db->metadata, txn->current_database, table_name) + 1) != 0) {
                    // Non-fatal error, data is already inserted
                }
                
                free(columns);
            } else {
                exec_result = GU_ERR_SYNTAX;
            }
            
        insert_cleanup:
            
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_CREATE_TABLE: {
            // Full CREATE TABLE implementation with metadata persistence
            if (stmt->data.create_table.table_name && txn->db->metadata) {
                char* table_name = stmt->data.create_table.table_name;
                
                // Count columns in the linked list and convert to column_info_t structures
                column_def_t* parsed_columns = stmt->data.create_table.columns;
                uint32_t column_count = 0;
                
                // First pass: count columns
                column_def_t* current = parsed_columns;
                while (current) {
                    column_count++;
                    current = current->next;
                }
                
                if (column_count > 0 && parsed_columns) {
                    // Check for duplicate column names
                    column_def_t* check_current = parsed_columns;
                    for (uint32_t i = 0; i < column_count && check_current; i++) {
                        column_def_t* check_next = check_current->next;
                        for (uint32_t j = i + 1; j < column_count && check_next; j++) {
                            if (strcmp(check_current->name, check_next->name) == 0) {
                                // Duplicate column name found
                                exec_result = GU_ERR_SYNTAX;
                                goto create_table_cleanup;
                            }
                            check_next = check_next->next;
                        }
                        check_current = check_current->next;
                    }
                    
                    column_info_t* columns = calloc(column_count, sizeof(column_info_t));
                    if (columns) {
                        // Convert each column definition
                        current = parsed_columns;
                        for (uint32_t i = 0; i < column_count && current; i++) {
                            strncpy(columns[i].column_name, current->name, sizeof(columns[i].column_name) - 1);
                            columns[i].data_type = current->type;
                            columns[i].max_length = current->length;
                            columns[i].is_primary_key = current->primary_key;
                            columns[i].is_unique = current->unique;
                            columns[i].is_not_null = current->not_null;
                            columns[i].is_auto_increment = current->auto_increment;
                            if (current->default_value) {
                                strncpy(columns[i].default_value, current->default_value, sizeof(columns[i].default_value) - 1);
                            }
                            current = current->next;
                        }
                        
                        // Add table to metadata manager (uses current database)
                        if (metadata_add_table(txn->db->metadata, txn->current_database, table_name, columns, column_count) == 0) {
                            // Table created successfully
                            
                            // Log to WAL
                            if (txn->db->wal_manager) {
                                uint32_t table_id = 1;
                                uint32_t page_id = 0;
                                uint64_t tuple_id = time(NULL);
                                
                                char log_msg[512];
                                snprintf(log_msg, sizeof(log_msg), "CREATE TABLE %s with %u columns", table_name, column_count);
                                
                                wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                             txn->xid, table_id, page_id, tuple_id,
                                             log_msg, strlen(log_msg));
                            }
                        } else {
                            exec_result = GU_ERR_IO;
                        }
                        
                        free(columns);
                    } else {
                        exec_result = GU_ERR_NOMEM;
                    }
                } else {
                    exec_result = GU_ERR_SYNTAX;
                }
                
            create_table_cleanup:
            } else {
                exec_result = GU_ERR_SYNTAX;
            }
            
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_CREATE_DATABASE: {
            // CREATE DATABASE implementation
            if (stmt->data.create_database.database_name && txn->db->metadata) {
                char* database_name = stmt->data.create_database.database_name;
                bool if_not_exists = stmt->data.create_database.if_not_exists;
                
                // Check if database already exists
                if (metadata_database_exists(txn->db->metadata, database_name)) {
                    if (if_not_exists) {
                        // Database exists but IF NOT EXISTS was specified, so succeed silently
                        res->row_count = 0;
                        res->col_count = 0;
                        break;
                    } else {
                        // Database exists and no IF NOT EXISTS, so error
                        exec_result = GU_ERR_SYNTAX; // Database already exists
                        break;
                    }
                }
                
                // Create the database
                if (metadata_add_database(txn->db->metadata, database_name) == 0) {
                    // Database created successfully
                    
                    // Log to WAL
                    if (txn->db->wal_manager) {
                        uint32_t table_id = 0; // Use 0 for database operations
                        uint32_t page_id = 0;  
                        uint64_t tuple_id = time(NULL);
                        
                        char log_msg[256];
                        snprintf(log_msg, sizeof(log_msg), "CREATE DATABASE %s", database_name);
                        
                        wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                     txn->xid, table_id, page_id, tuple_id,
                                     log_msg, strlen(log_msg));
                    }
                    
                    res->row_count = 1; // Indicate one database was created
                    res->col_count = 0;
                } else {
                    exec_result = GU_ERR_IO; // Failed to create database
                }
            } else {
                exec_result = GU_ERR_SYNTAX; // Missing database name or metadata
            }
            break;
        }
        
        case STMT_DROP_DATABASE: {
            // DROP DATABASE implementation
            if (stmt->data.drop_database.database_name && txn->db->metadata) {
                char* database_name = stmt->data.drop_database.database_name;
                bool if_exists = stmt->data.drop_database.if_exists;
                
                // Check if database exists
                if (!metadata_database_exists(txn->db->metadata, database_name)) {
                    if (if_exists) {
                        // Database doesn't exist but IF EXISTS was specified, so succeed silently
                        res->row_count = 0;
                        res->col_count = 0;
                        break;
                    } else {
                        // Database doesn't exist and no IF EXISTS, so error
                        exec_result = GU_ERR_SYNTAX; // Database doesn't exist
                        break;
                    }
                }
                
                // Don't allow dropping the current database
                if (strcmp(txn->current_database, database_name) == 0) {
                    exec_result = GU_ERR_SYNTAX; // Cannot drop current database
                    break;
                }
                
                // Remove the database
                if (metadata_remove_database(txn->db->metadata, database_name) == 0) {
                    // Database dropped successfully
                    
                    // Log to WAL
                    if (txn->db->wal_manager) {
                        uint32_t table_id = 0; // Use 0 for database operations
                        uint32_t page_id = 0;  
                        uint64_t tuple_id = time(NULL);
                        
                        char log_msg[256];
                        snprintf(log_msg, sizeof(log_msg), "DROP DATABASE %s", database_name);
                        
                        wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                     txn->xid, table_id, page_id, tuple_id,
                                     log_msg, strlen(log_msg));
                    }
                    
                    res->row_count = 1; // Indicate one database was dropped
                    res->col_count = 0;
                } else {
                    exec_result = GU_ERR_IO; // Failed to drop database
                }
            } else {
                exec_result = GU_ERR_SYNTAX; // Missing database name or metadata
            }
            break;
        }
        
        case STMT_DROP_TABLE: {
            // DROP TABLE implementation
            if (stmt->data.drop_table.table_name && txn->db->metadata) {
                char* table_name = stmt->data.drop_table.table_name;
                bool if_exists = stmt->data.drop_table.if_exists;
                
                // Check if table exists in current database
                if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                    if (if_exists) {
                        // Table doesn't exist but IF EXISTS was specified, so succeed silently
                        res->row_count = 0;
                        res->col_count = 0;
                        break;
                    } else {
                        // Table doesn't exist and no IF EXISTS, so error
                        exec_result = GU_ERR_SYNTAX; // Table doesn't exist
                        break;
                    }
                }
                
                // Remove the table
                if (metadata_remove_table(txn->db->metadata, txn->current_database, table_name) == 0) {
                    // Table dropped successfully
                    
                    // Log to WAL
                    if (txn->db->wal_manager) {
                        uint32_t table_id = 1; // Use 1 for table operations
                        uint32_t page_id = 0;  
                        uint64_t tuple_id = time(NULL);
                        
                        char log_msg[256];
                        snprintf(log_msg, sizeof(log_msg), "DROP TABLE %s.%s", txn->current_database, table_name);
                        
                        wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                     txn->xid, table_id, page_id, tuple_id,
                                     log_msg, strlen(log_msg));
                    }
                    
                    res->row_count = 1; // Indicate one table was dropped
                    res->col_count = 0;
                } else {
                    exec_result = GU_ERR_IO; // Failed to drop table
                }
            } else {
                exec_result = GU_ERR_SYNTAX; // Missing table name or metadata
            }
            break;
        }
        
        case STMT_UPDATE: {
            // UPDATE implementation
            if (!stmt->data.update.table_name || !stmt->data.update.assignments || stmt->data.update.assignment_count == 0) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* table_name = stmt->data.update.table_name;
            
            // Check if table exists in current database
            if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                exec_result = GU_ERR_SYNTAX; // Table not found
                break;
            }
            
            // Get table column information for validation
            column_info_t* columns = NULL;
            uint32_t column_count = 0;
            if (metadata_get_table_columns(txn->db->metadata, txn->current_database, table_name, &columns, &column_count) != 0 || !columns) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            // Validate assignment columns exist in table
            for (uint32_t i = 0; i < stmt->data.update.assignment_count; i++) {
                expression_t* assignment = stmt->data.update.assignments[i];
                if (assignment->type == EXPR_BINARY_OP && assignment->data.binary_op.left &&
                    assignment->data.binary_op.left->type == EXPR_COLUMN) {
                    
                    char* column_name = assignment->data.binary_op.left->data.column.column_name;
                    bool column_found = false;
                    
                    for (uint32_t j = 0; j < column_count; j++) {
                        if (strcmp(columns[j].column_name, column_name) == 0) {
                            column_found = true;
                            break;
                        }
                    }
                    
                    if (!column_found) {
                        free(columns);
                        exec_result = GU_ERR_SYNTAX; // Column not found
                        break;
                    }
                }
            }
            
            if (exec_result != GU_OK) {
                if (columns) free(columns);
                break;
            }
            
            // REAL UPDATE implementation
            // 1. Scan storage to find matching rows (simplified: update all rows if no WHERE clause)
            // 2. Apply assignments to each matching row
            // 3. Update storage
            // 4. Log to WAL
            
            uint32_t updated_rows = 0;
            uint32_t max_rows_to_check = 1000;
            
            for (uint32_t row_id = 1; row_id <= max_rows_to_check; row_id++) {
                char search_key[256];
                snprintf(search_key, sizeof(search_key), "%s:%u", table_name, row_id);
                
                search_result_t search_result = storage_search((bw_tree_t*)txn->db->storage, 
                                                            search_key, strlen(search_key));
                if (search_result.found && search_result.value) {
                    // Check WHERE clause condition if present
                    bool should_update = true;
                    if (stmt->data.update.where_clause) {
                        // Create tuple from JSON data for WHERE evaluation
                        tuple_t* tuple = create_tuple_from_json(search_result.value, columns, column_count);
                        if (tuple) {
                            // Create executor context for expression evaluation
                            executor_context_t* exec_context = malloc(sizeof(executor_context_t));
                            exec_context->db = txn->db;
                            exec_context->transaction = txn;
                            
                            // Evaluate WHERE clause
                            should_update = evaluate_qualification(stmt->data.update.where_clause, tuple, exec_context);
                            
                            // Clean up
                            tuple_destroy(tuple);
                            free(exec_context);
                        } else {
                            should_update = false; // Failed to parse tuple
                        }
                    }
                    
                    if (!should_update) {
                        continue; // Skip this row
                    }
                    
                    // Parse existing JSON data
                    char updated_value[2048];
                    strcpy(updated_value, search_result.value); // Start with existing data
                    
                    // Apply each assignment
                    for (uint32_t i = 0; i < stmt->data.update.assignment_count; i++) {
                        expression_t* assignment = stmt->data.update.assignments[i];
                        if (assignment->type == EXPR_BINARY_OP && 
                            assignment->data.binary_op.left &&
                            assignment->data.binary_op.left->type == EXPR_COLUMN &&
                            assignment->data.binary_op.right &&
                            assignment->data.binary_op.right->type == EXPR_LITERAL) {
                            
                            char* column_name = assignment->data.binary_op.left->data.column.column_name;
                            char* new_value = assignment->data.binary_op.right->data.literal.value;
                            
                            // Find and replace the column value in JSON
                            char old_pattern[256];
                            snprintf(old_pattern, sizeof(old_pattern), "\"%s\":\"", column_name);
                            
                            char* start = strstr(updated_value, old_pattern);
                            if (start) {
                                start += strlen(old_pattern);
                                char* end = strchr(start, '"');
                                if (end) {
                                    // Calculate lengths for replacement
                                    size_t prefix_len = start - updated_value;
                                    size_t suffix_len = strlen(end);
                                    
                                    // Build new JSON string
                                    char temp_value[2048];
                                    strncpy(temp_value, updated_value, prefix_len);
                                    temp_value[prefix_len] = '\0';
                                    
                                    strcat(temp_value, new_value);
                                    strcat(temp_value, end);
                                    
                                    strcpy(updated_value, temp_value);
                                }
                            }
                        }
                    }
                    
                    // Update storage with new value
                    int update_result = storage_insert((bw_tree_t*)txn->db->storage, 
                                                     search_key, strlen(search_key),
                                                     updated_value, strlen(updated_value));
                    if (update_result == 0) {
                        updated_rows++;
                        
                        // Log to WAL
                        if (txn->db->wal_manager) {
                            uint32_t table_id = 1;
                            uint32_t page_id = 0;
                            uint64_t tuple_id = (uint64_t)row_id;
                            
                            wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                         txn->xid, table_id, page_id, tuple_id,
                                         updated_value, strlen(updated_value));
                        }
                    }
                }
            }
            
            res->row_count = updated_rows; // Return actual number of updated rows
            res->col_count = 0;
            
            free(columns);
            break;
        }
        
        case STMT_DELETE: {
            // REAL DELETE implementation
            if (!stmt->data.delete.table_name) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* table_name = stmt->data.delete.table_name;
            
            // Check if table exists in current database
            if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                exec_result = GU_ERR_SYNTAX; // Table not found
                break;
            }
            
            // Get table column information for validation
            column_info_t* columns = NULL;
            uint32_t column_count = 0;
            if (metadata_get_table_columns(txn->db->metadata, txn->current_database, table_name, 
                                         &columns, &column_count) != 0 || !columns) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            // Scan storage to find and delete matching rows
            uint32_t deleted_rows = 0;
            uint32_t max_rows_to_check = 1000;
            
            for (uint32_t row_id = 1; row_id <= max_rows_to_check; row_id++) {
                char search_key[256];
                snprintf(search_key, sizeof(search_key), "%s:%u", table_name, row_id);
                
                search_result_t search_result = storage_search((bw_tree_t*)txn->db->storage, 
                                                            search_key, strlen(search_key));
                if (search_result.found) {
                    // Check WHERE clause condition if present
                    bool should_delete = true;
                    if (stmt->data.delete.where_clause) {
                        // Get table column information for WHERE evaluation
                        column_info_t* delete_columns = NULL;
                        uint32_t delete_column_count = 0;
                        if (metadata_get_table_columns(txn->db->metadata, txn->current_database, table_name, &delete_columns, &delete_column_count) == 0 && delete_columns) {
                            // Create tuple from JSON data for WHERE evaluation
                            tuple_t* tuple = create_tuple_from_json(search_result.value, delete_columns, delete_column_count);
                            if (tuple) {
                                // Create executor context for expression evaluation
                                executor_context_t* exec_context = malloc(sizeof(executor_context_t));
                                exec_context->db = txn->db;
                                exec_context->transaction = txn;
                                
                                // Evaluate WHERE clause
                                should_delete = evaluate_qualification(stmt->data.delete.where_clause, tuple, exec_context);
                                
                                // Clean up
                                tuple_destroy(tuple);
                                free(exec_context);
                            } else {
                                should_delete = false; // Failed to parse tuple
                            }
                            free(delete_columns);
                        } else {
                            should_delete = false; // Failed to get column info
                        }
                    }
                    
                    if (!should_delete) {
                        continue; // Skip this row
                    }
                    
                    // Delete from storage
                    int delete_result = storage_delete((bw_tree_t*)txn->db->storage, 
                                                     search_key, strlen(search_key));
                    if (delete_result == 0) {
                        deleted_rows++;
                        
                        // Log to WAL
                        if (txn->db->wal_manager) {
                            uint32_t table_id = 1;
                            uint32_t page_id = 0;
                            uint64_t tuple_id = (uint64_t)row_id;
                            
                            char log_msg[256];
                            snprintf(log_msg, sizeof(log_msg), "DELETE FROM %s.%s WHERE row_id=%u", 
                                   txn->current_database, table_name, row_id);
                            
                            wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                         txn->xid, table_id, page_id, tuple_id,
                                         log_msg, strlen(log_msg));
                        }
                    }
                }
            }
            
            // Update row count in metadata
            if (deleted_rows > 0) {
                uint64_t current_count = metadata_get_table_row_count(txn->db->metadata, 
                                                                    txn->current_database, table_name);
                uint64_t new_count = (current_count > deleted_rows) ? (current_count - deleted_rows) : 0;
                metadata_update_row_count(txn->db->metadata, txn->current_database, table_name, new_count);
            }
            
            res->row_count = deleted_rows; // Return actual number of deleted rows
            res->col_count = 0;
            
            free(columns);
            break;
        }
        
        case STMT_USE: {
            // Implement USE database switching
            if (!stmt->data.use.database_name) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* database_name = stmt->data.use.database_name;
            
            // Check if database exists
            if (!metadata_database_exists(txn->db->metadata, database_name)) {
                exec_result = GU_ERR_SYNTAX; // Database doesn't exist
                break;
            }
            
            // Switch current database
            strncpy(txn->current_database, database_name, sizeof(txn->current_database) - 1);
            txn->current_database[sizeof(txn->current_database) - 1] = '\0';
            
            // Mark that user has explicitly selected a database
            txn->database_explicitly_selected = true;
            
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_DESCRIBE: {
            // DESCRIBE table - show table structure (same as SHOW COLUMNS)
            if (!stmt->data.describe.table_name) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* table_name = stmt->data.describe.table_name;
            
            // Check if table exists
            if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                exec_result = GU_ERR_SYNTAX; // Table not found
                break;
            }
            
            // Get table column information
            column_info_t* columns = NULL;
            uint32_t column_count = 0;
            if (metadata_get_table_columns(txn->db->metadata, txn->current_database, table_name, &columns, &column_count) != 0 || !columns) {
                exec_result = GU_ERR_IO;
                break;
            }
            
            // Set up result with column information (Field, Type, Null, Key, Default, Extra)
            res->col_count = 6;
            res->column_names = calloc(6, sizeof(char*));
            res->column_names[0] = strdup("Field");
            res->column_names[1] = strdup("Type");
            res->column_names[2] = strdup("Null");
            res->column_names[3] = strdup("Key");
            res->column_names[4] = strdup("Default");
            res->column_names[5] = strdup("Extra");
            
            res->row_count = column_count;
            res->rows = calloc(column_count, sizeof(char**));
            
            for (uint32_t i = 0; i < column_count; i++) {
                res->rows[i] = calloc(6, sizeof(char*));
                res->rows[i][0] = strdup(columns[i].column_name);
                // Convert data_type (uint32_t) to string
                const char* type_str = data_type_to_string(columns[i].data_type);
                res->rows[i][1] = strdup(type_str ? type_str : "UNKNOWN");
                res->rows[i][2] = strdup(columns[i].is_not_null ? "NO" : "YES");
                res->rows[i][3] = strdup(columns[i].is_primary_key ? "PRI" : "");
                res->rows[i][4] = strdup(strlen(columns[i].default_value) > 0 ? columns[i].default_value : "NULL");
                res->rows[i][5] = strdup(columns[i].is_auto_increment ? "auto_increment" : "");
            }
            
            free(columns);
            break;
        }
        
        case STMT_SHOW_DATABASES: {
            // Already handled by handle_show_command for "show databases"
            // But we can implement it here too for consistency
            res->col_count = 1;
            res->column_names = calloc(1, sizeof(char*));
            res->column_names[0] = strdup("Database");
            
            if (txn->db->metadata) {
                database_info_t* databases = NULL;
                uint32_t db_count = 0;
                
                if (metadata_list_databases(txn->db->metadata, &databases, &db_count) == 0 && db_count > 0) {
                    res->row_count = db_count;
                    res->rows = calloc(db_count, sizeof(char**));
                    
                    for (uint32_t i = 0; i < db_count; i++) {
                        res->rows[i] = calloc(1, sizeof(char*));
                        res->rows[i][0] = strdup(databases[i].name);
                    }
                    
                    free(databases);
                } else {
                    res->row_count = 0;
                    res->rows = NULL;
                }
            } else {
                res->row_count = 0;
                res->rows = NULL;
            }
            break;
        }
        
        case STMT_SHOW_TABLES: {
            // Check if user has explicitly selected a database
            if (!txn->database_explicitly_selected) {
                exec_result = GU_ERR_SYNTAX;
                res->error_msg = strdup("No database selected. Use 'USE database_name' first.");
                break;
            }
            
            // Show tables for current database
            res->col_count = 1;
            res->column_names = calloc(1, sizeof(char*));
            
            // Dynamic column name based on current database
            char column_name[128];
            snprintf(column_name, sizeof(column_name), "Tables_in_%s", txn->current_database);
            res->column_names[0] = strdup(column_name);
            
            if (txn->db->metadata) {
                table_info_t* tables = NULL;
                uint32_t table_count = 0;
                
                if (metadata_list_tables(txn->db->metadata, txn->current_database, &tables, &table_count) == 0 && table_count > 0) {
                    res->row_count = table_count;
                    res->rows = calloc(table_count, sizeof(char**));
                    
                    for (uint32_t i = 0; i < table_count; i++) {
                        res->rows[i] = calloc(1, sizeof(char*));
                        res->rows[i][0] = strdup(tables[i].table_name);
                    }
                    
                    free(tables);
                } else {
                    res->row_count = 0;
                    res->rows = NULL;
                }
            } else {
                res->row_count = 0;
                res->rows = NULL;
            }
            break;
        }
        
        case STMT_SHOW_COLUMNS: {
            // SHOW COLUMNS FROM table (same as DESCRIBE)
            if (!stmt->data.show.table_name) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* table_name = stmt->data.show.table_name;
            
            // Check if table exists
            if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                exec_result = GU_ERR_SYNTAX; // Table not found
                break;
            }
            
            // Get table column information
            column_info_t* columns = NULL;
            uint32_t column_count = 0;
            if (metadata_get_table_columns(txn->db->metadata, txn->current_database, table_name, &columns, &column_count) != 0 || !columns) {
                exec_result = GU_ERR_IO;
                break;
            }
            
            // Set up result with column information
            res->col_count = 6;
            res->column_names = calloc(6, sizeof(char*));
            res->column_names[0] = strdup("Field");
            res->column_names[1] = strdup("Type");
            res->column_names[2] = strdup("Null");
            res->column_names[3] = strdup("Key");
            res->column_names[4] = strdup("Default");
            res->column_names[5] = strdup("Extra");
            
            res->row_count = column_count;
            res->rows = calloc(column_count, sizeof(char**));
            
            for (uint32_t i = 0; i < column_count; i++) {
                res->rows[i] = calloc(6, sizeof(char*));
                res->rows[i][0] = strdup(columns[i].column_name);
                // Convert data_type (uint32_t) to string
                const char* type_str = data_type_to_string(columns[i].data_type);
                res->rows[i][1] = strdup(type_str ? type_str : "UNKNOWN");
                res->rows[i][2] = strdup(columns[i].is_not_null ? "NO" : "YES");
                res->rows[i][3] = strdup(columns[i].is_primary_key ? "PRI" : "");
                res->rows[i][4] = strdup(strlen(columns[i].default_value) > 0 ? columns[i].default_value : "NULL");
                res->rows[i][5] = strdup(columns[i].is_auto_increment ? "auto_increment" : "");
            }
            
            free(columns);
            break;
        }
        
        case STMT_SHOW_CREATE_TABLE: {
            // SHOW CREATE TABLE - return the CREATE TABLE statement
            if (!stmt->data.show.table_name) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* table_name = stmt->data.show.table_name;
            
            // Check if table exists
            if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                exec_result = GU_ERR_SYNTAX; // Table not found
                break;
            }
            
            // Set up result with Table and Create Table columns
            res->col_count = 2;
            res->column_names = calloc(2, sizeof(char*));
            res->column_names[0] = strdup("Table");
            res->column_names[1] = strdup("Create Table");
            
            res->row_count = 1;
            res->rows = calloc(1, sizeof(char**));
            res->rows[0] = calloc(2, sizeof(char*));
            res->rows[0][0] = strdup(table_name);
            
            // Generate CREATE TABLE statement from metadata
            column_info_t* table_columns = NULL;
            uint32_t table_column_count = 0;
            if (metadata_get_table_columns(txn->db->metadata, txn->current_database, table_name, &table_columns, &table_column_count) == 0 && table_columns) {
                // Build CREATE TABLE statement
                char create_table_sql[4096];
                snprintf(create_table_sql, sizeof(create_table_sql), "CREATE TABLE `%s` (\n  ", table_name);
                
                // Add each column definition
                for (uint32_t i = 0; i < table_column_count; i++) {
                    char column_def[512];
                    char data_type_str[64];
                    
                    // Convert data_type to string
                    switch (table_columns[i].data_type) {
                        case DATA_TYPE_TINYINT:
                            strcpy(data_type_str, "TINYINT");
                            break;
                        case DATA_TYPE_INT:
                            strcpy(data_type_str, "INT");
                            break;
                        case DATA_TYPE_BIGINT:
                            strcpy(data_type_str, "BIGINT");
                            break;
                        case DATA_TYPE_FLOAT:
                            strcpy(data_type_str, "FLOAT");
                            break;
                        case DATA_TYPE_DOUBLE:
                            strcpy(data_type_str, "DOUBLE");
                            break;
                        case DATA_TYPE_CHAR:
                            snprintf(data_type_str, sizeof(data_type_str), "CHAR(%u)", table_columns[i].max_length);
                            break;
                        case DATA_TYPE_VARCHAR:
                            snprintf(data_type_str, sizeof(data_type_str), "VARCHAR(%u)", table_columns[i].max_length);
                            break;
                        case DATA_TYPE_TEXT:
                            strcpy(data_type_str, "TEXT");
                            break;
                        case DATA_TYPE_TIMESTAMP:
                            strcpy(data_type_str, "TIMESTAMP");
                            break;
                        default:
                            strcpy(data_type_str, "UNKNOWN");
                            break;
                    }
                    
                    // Build column definition
                    snprintf(column_def, sizeof(column_def), "`%s` %s", table_columns[i].column_name, data_type_str);
                    
                    // Add constraints
                    if (table_columns[i].is_not_null) {
                        strcat(column_def, " NOT NULL");
                    }
                    if (table_columns[i].is_auto_increment) {
                        strcat(column_def, " AUTO_INCREMENT");
                    }
                    if (strlen(table_columns[i].default_value) > 0) {
                        char default_clause[128];
                        snprintf(default_clause, sizeof(default_clause), " DEFAULT '%s'", table_columns[i].default_value);
                        strcat(column_def, default_clause);
                    }
                    
                    // Add column to SQL
                    strcat(create_table_sql, column_def);
                    
                    // Add comma and newline for next column
                    if (i < table_column_count - 1) {
                        strcat(create_table_sql, ",\n  ");
                    }
                }
                
                // Add primary key constraint if any
                bool has_primary_key = false;
                for (uint32_t i = 0; i < table_column_count; i++) {
                    if (table_columns[i].is_primary_key) {
                        if (!has_primary_key) {
                            strcat(create_table_sql, ",\n  PRIMARY KEY (");
                            has_primary_key = true;
                        } else {
                            strcat(create_table_sql, ", ");
                        }
                        char pk_column[128];
                        snprintf(pk_column, sizeof(pk_column), "`%s`", table_columns[i].column_name);
                        strcat(create_table_sql, pk_column);
                    }
                }
                if (has_primary_key) {
                    strcat(create_table_sql, ")");
                }
                
                // Close the statement
                strcat(create_table_sql, "\n)");
                
                res->rows[0][1] = strdup(create_table_sql);
                free(table_columns);
            } else {
                res->rows[0][1] = strdup("CREATE TABLE statement reconstruction failed");
            }
            
            break;
        }
        
        case STMT_SHOW_STATUS: {
            // SHOW STATUS - return server status
            res->col_count = 2;
            res->column_names = calloc(2, sizeof(char*));
            res->column_names[0] = strdup("Variable_name");
            res->column_names[1] = strdup("Value");
            
            // Add some basic status variables
            res->row_count = 3;
            res->rows = calloc(3, sizeof(char**));
            
            res->rows[0] = calloc(2, sizeof(char*));
            res->rows[0][0] = strdup("Uptime");
            res->rows[0][1] = strdup("Unknown");
            
            res->rows[1] = calloc(2, sizeof(char*));
            res->rows[1][0] = strdup("Threads_connected");
            res->rows[1][1] = strdup("1");
            
            res->rows[2] = calloc(2, sizeof(char*));
            res->rows[2][0] = strdup("Questions");
            res->rows[2][1] = strdup("0");
            
            break;
        }
        
        case STMT_ALTER_TABLE: {
            if (!stmt->data.alter_table.table_name || !txn->current_database) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* table_name = stmt->data.alter_table.table_name;
            
            // Check if table exists in current database
            if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                exec_result = GU_ERR_SYNTAX; // Table doesn't exist
                break;
            }
            
            // Handle different ALTER TABLE operations
            switch (stmt->data.alter_table.operation) {
                case ALTER_ADD_COLUMN: {
                    if (!stmt->data.alter_table.column_name) {
                        exec_result = GU_ERR_SYNTAX;
                        break;
                    }
                    
                    char* column_name = stmt->data.alter_table.column_name;
                    
                    // Check if column already exists
                    column_info_t* existing_columns = NULL;
                    uint32_t existing_column_count = 0;
                    if (metadata_get_table_columns(txn->db->metadata, txn->current_database, table_name, 
                                                 &existing_columns, &existing_column_count) == 0) {
                        
                        // Check for duplicate column name
                        for (uint32_t i = 0; i < existing_column_count; i++) {
                            if (strcmp(existing_columns[i].column_name, column_name) == 0) {
                                free(existing_columns);
                                exec_result = GU_ERR_SYNTAX; // Column already exists
                                goto alter_table_end;
                            }
                        }
                        free(existing_columns);
                    }
                    
                    // Create new column info
                    column_info_t new_column;
                    memset(&new_column, 0, sizeof(column_info_t));
                    strncpy(new_column.database_name, txn->current_database, sizeof(new_column.database_name) - 1);
                    strncpy(new_column.table_name, table_name, sizeof(new_column.table_name) - 1);
                    strncpy(new_column.column_name, column_name, sizeof(new_column.column_name) - 1);
                    new_column.data_type = stmt->data.alter_table.column_type;
                    new_column.max_length = stmt->data.alter_table.column_length;
                    new_column.is_primary_key = stmt->data.alter_table.primary_key;
                    new_column.is_unique = stmt->data.alter_table.unique;
                    new_column.is_not_null = stmt->data.alter_table.not_null;
                    if (stmt->data.alter_table.default_value) {
                        strncpy(new_column.default_value, stmt->data.alter_table.default_value, 
                               sizeof(new_column.default_value) - 1);
                    }
                    
                    // Add column to metadata
                    if (metadata_add_column(txn->db->metadata, txn->current_database, table_name, &new_column) == 0) {
                        // Log to WAL
                        if (txn->db->wal_manager) {
                            char log_msg[512];
                            snprintf(log_msg, sizeof(log_msg), "ALTER TABLE %s ADD COLUMN %s", 
                                    table_name, column_name);
                            
                            wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                         txn->xid, 1, 0, time(NULL),
                                         log_msg, strlen(log_msg));
                        }
                    } else {
                        exec_result = GU_ERR_IO;
                    }
                    break;
                }
                
                case ALTER_DROP_COLUMN: {
                    if (!stmt->data.alter_table.column_name) {
                        exec_result = GU_ERR_SYNTAX;
                        break;
                    }
                    
                    char* column_name = stmt->data.alter_table.column_name;
                    
                    // Remove column from metadata
                    if (metadata_remove_column(txn->db->metadata, txn->current_database, table_name, column_name) == 0) {
                        // Log to WAL
                        if (txn->db->wal_manager) {
                            char log_msg[512];
                            snprintf(log_msg, sizeof(log_msg), "ALTER TABLE %s DROP COLUMN %s", 
                                    table_name, column_name);
                            
                            wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                         txn->xid, 1, 0, time(NULL),
                                         log_msg, strlen(log_msg));
                        }
                    } else {
                        exec_result = GU_ERR_SYNTAX; // Column doesn't exist or other error
                    }
                    break;
                }
                
                case ALTER_MODIFY_COLUMN: {
                    if (!stmt->data.alter_table.column_name) {
                        exec_result = GU_ERR_SYNTAX;
                        break;
                    }
                    
                    char* column_name = stmt->data.alter_table.column_name;
                    
                    // Create new column info for modification
                    column_info_t new_column_info;
                    memset(&new_column_info, 0, sizeof(column_info_t));
                    strncpy(new_column_info.database_name, txn->current_database, sizeof(new_column_info.database_name) - 1);
                    strncpy(new_column_info.table_name, table_name, sizeof(new_column_info.table_name) - 1);
                    strncpy(new_column_info.column_name, column_name, sizeof(new_column_info.column_name) - 1);
                    new_column_info.data_type = stmt->data.alter_table.column_type;
                    new_column_info.max_length = stmt->data.alter_table.column_length;
                    new_column_info.is_primary_key = stmt->data.alter_table.primary_key;
                    new_column_info.is_unique = stmt->data.alter_table.unique;
                    new_column_info.is_not_null = stmt->data.alter_table.not_null;
                    if (stmt->data.alter_table.default_value) {
                        strncpy(new_column_info.default_value, stmt->data.alter_table.default_value, 
                               sizeof(new_column_info.default_value) - 1);
                    }
                    
                    // Modify column in metadata
                    if (metadata_modify_column(txn->db->metadata, txn->current_database, table_name, 
                                             column_name, &new_column_info) == 0) {
                        // Log to WAL
                        if (txn->db->wal_manager) {
                            char log_msg[512];
                            snprintf(log_msg, sizeof(log_msg), "ALTER TABLE %s MODIFY COLUMN %s", 
                                    table_name, column_name);
                            
                            wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                         txn->xid, 1, 0, time(NULL),
                                         log_msg, strlen(log_msg));
                        }
                    } else {
                        exec_result = GU_ERR_SYNTAX; // Column doesn't exist or other error
                    }
                    break;
                }
                
                case ALTER_RENAME_COLUMN: {
                    if (!stmt->data.alter_table.column_name || !stmt->data.alter_table.new_column_name) {
                        exec_result = GU_ERR_SYNTAX;
                        break;
                    }
                    
                    char* old_column_name = stmt->data.alter_table.column_name;
                    char* new_column_name = stmt->data.alter_table.new_column_name;
                    
                    // Rename column in metadata
                    if (metadata_rename_column(txn->db->metadata, txn->current_database, table_name, 
                                             old_column_name, new_column_name) == 0) {
                        // Log to WAL
                        if (txn->db->wal_manager) {
                            char log_msg[512];
                            snprintf(log_msg, sizeof(log_msg), "ALTER TABLE %s RENAME COLUMN %s TO %s", 
                                    table_name, old_column_name, new_column_name);
                            
                            wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                         txn->xid, 1, 0, time(NULL),
                                         log_msg, strlen(log_msg));
                        }
                    } else {
                        exec_result = GU_ERR_SYNTAX; // Column doesn't exist or new name conflicts
                    }
                    break;
                }
                
                default:
                    exec_result = GU_ERR_SYNTAX; // Unsupported operation
                    break;
            }
            
            alter_table_end:
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_CREATE_INDEX: {
            if (!stmt->data.create_index.index_name || !stmt->data.create_index.table_name || 
                !stmt->data.create_index.column_names || !txn->current_database) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* index_name = stmt->data.create_index.index_name;
            char* table_name = stmt->data.create_index.table_name;
            
            // Check if table exists in current database
            if (!metadata_table_exists(txn->db->metadata, txn->current_database, table_name)) {
                exec_result = GU_ERR_SYNTAX; // Table doesn't exist
                break;
            }
            
            // Check if index already exists
            if (metadata_index_exists(txn->db->metadata, txn->current_database, index_name)) {
                if (stmt->data.create_index.if_not_exists) {
                    // If IF NOT EXISTS is specified, just succeed silently
                    exec_result = GU_OK;
                } else {
                    exec_result = GU_ERR_SYNTAX; // Index already exists
                }
                break;
            }
            
            // Create index info
            index_info_t new_index;
            memset(&new_index, 0, sizeof(index_info_t));
            strncpy(new_index.database_name, txn->current_database, sizeof(new_index.database_name) - 1);
            strncpy(new_index.table_name, table_name, sizeof(new_index.table_name) - 1);
            strncpy(new_index.index_name, index_name, sizeof(new_index.index_name) - 1);
            
            // Build comma-separated column names string
            char column_names_str[512] = {0};
            for (uint32_t i = 0; i < stmt->data.create_index.column_count; i++) {
                if (i > 0) {
                    strncat(column_names_str, ",", sizeof(column_names_str) - strlen(column_names_str) - 1);
                }
                strncat(column_names_str, stmt->data.create_index.column_names[i], 
                       sizeof(column_names_str) - strlen(column_names_str) - 1);
            }
            strncpy(new_index.column_names, column_names_str, sizeof(new_index.column_names) - 1);
            
            new_index.column_count = stmt->data.create_index.column_count;
            new_index.is_unique = stmt->data.create_index.unique;
            new_index.is_primary = false; // CREATE INDEX doesn't create primary indexes
            new_index.created_time = time(NULL);
            new_index.is_active = true;
            
            // Add index to metadata
            if (metadata_add_index(txn->db->metadata, txn->current_database, table_name, &new_index) == 0) {
                // Log to WAL
                if (txn->db->wal_manager) {
                    char log_msg[512];
                    snprintf(log_msg, sizeof(log_msg), "CREATE INDEX %s ON %s (%s)", 
                            index_name, table_name, column_names_str);
                    
                    wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                 txn->xid, 1, 0, time(NULL),
                                 log_msg, strlen(log_msg));
                }
            } else {
                exec_result = GU_ERR_IO;
            }
            
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_DROP_INDEX: {
            if (!stmt->data.drop_index.index_name || !txn->current_database) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* index_name = stmt->data.drop_index.index_name;
            
            // Check if index exists
            if (!metadata_index_exists(txn->db->metadata, txn->current_database, index_name)) {
                if (stmt->data.drop_index.if_exists) {
                    // If IF EXISTS is specified, just succeed silently
                    exec_result = GU_OK;
                } else {
                    exec_result = GU_ERR_SYNTAX; // Index doesn't exist
                }
                break;
            }
            
            // Remove index from metadata
            if (metadata_remove_index(txn->db->metadata, txn->current_database, index_name) == 0) {
                // Log to WAL
                if (txn->db->wal_manager) {
                    char log_msg[512];
                    snprintf(log_msg, sizeof(log_msg), "DROP INDEX %s", index_name);
                    
                    wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                 txn->xid, 1, 0, time(NULL),
                                 log_msg, strlen(log_msg));
                }
            } else {
                exec_result = GU_ERR_SYNTAX; // Error removing index
            }
            
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_CREATE_VIEW: {
            if (!stmt->data.create_view.view_name || !txn->current_database) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* view_name = stmt->data.create_view.view_name;
            
            // Create view_info_t structure
            view_info_t view_info;
            memset(&view_info, 0, sizeof(view_info_t));
            strncpy(view_info.view_name, view_name, sizeof(view_info.view_name) - 1);
            view_info.or_replace = stmt->data.create_view.or_replace;
            
            // Build SELECT SQL from parsed components (simplified)
            char select_sql[2048];
            snprintf(select_sql, sizeof(select_sql), "SELECT * FROM %s", 
                    stmt->data.create_view.from_list ? stmt->data.create_view.from_list->table_name : "unknown");
            strncpy(view_info.select_sql, select_sql, sizeof(view_info.select_sql) - 1);
            
            // Build column names from select_list
            if (stmt->data.create_view.select_list) {
                char column_names_str[512] = "";
                uint32_t column_count = 0;
                
                // Handle different expression types in SELECT list
                expression_t* expr = stmt->data.create_view.select_list;
                
                if (expr->type == EXPR_LIST) {
                    // Multiple columns in SELECT list
                    for (uint32_t i = 0; i < expr->data.list.count; i++) {
                        expression_t* col_expr = expr->data.list.expressions[i];
                        
                        if (col_expr->type == EXPR_COLUMN) {
                            // Column reference: table.column or just column
                            if (column_count > 0) {
                                strcat(column_names_str, ",");
                            }
                            if (col_expr->data.column.column_name) {
                                strcat(column_names_str, col_expr->data.column.column_name);
                            } else {
                                strcat(column_names_str, "unknown_column");
                            }
                            column_count++;
                        } else if (col_expr->type == EXPR_STAR) {
                            // SELECT * case
                            strcpy(column_names_str, "*");
                            column_count = 1;
                            break;
                        } else if (col_expr->type == EXPR_FUNCTION_CALL) {
                            // Function call like COUNT(*), SUM(column)
                            if (column_count > 0) {
                                strcat(column_names_str, ",");
                            }
                            if (col_expr->data.function_call.function_name) {
                                strcat(column_names_str, col_expr->data.function_call.function_name);
                            } else {
                                strcat(column_names_str, "function_result");
                            }
                            column_count++;
                        } else {
                            // Other expression types (literals, etc.)
                            if (column_count > 0) {
                                strcat(column_names_str, ",");
                            }
                            strcat(column_names_str, "expr_result");
                            column_count++;
                        }
                    }
                } else if (expr->type == EXPR_STAR) {
                    // Simple SELECT *
                    strcpy(column_names_str, "*");
                    column_count = 1;
                } else if (expr->type == EXPR_COLUMN) {
                    // Single column
                    if (expr->data.column.column_name) {
                        strcpy(column_names_str, expr->data.column.column_name);
                    } else {
                        strcpy(column_names_str, "unknown_column");
                    }
                    column_count = 1;
                } else {
                    // Default fallback
                    strcpy(column_names_str, "result");
                    column_count = 1;
                }
                
                strncpy(view_info.column_names, column_names_str, sizeof(view_info.column_names) - 1);
                view_info.column_count = column_count;
            } else {
                // Fallback if no select_list
                strcpy(view_info.column_names, "*");
                view_info.column_count = 1;
            }
            
            // Add view to metadata
            if (metadata_add_view(txn->db->metadata, txn->current_database, &view_info) == 0) {
                // Log to WAL
                if (txn->db->wal_manager) {
                    char log_msg[512];
                    snprintf(log_msg, sizeof(log_msg), "CREATE VIEW %s AS %s", 
                            view_name, select_sql);
                    
                    wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                 txn->xid, 1, 0, time(NULL),
                                 log_msg, strlen(log_msg));
                }
                exec_result = GU_OK;
            } else {
                exec_result = GU_ERR_SYNTAX; // View creation failed
            }
            
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_DROP_VIEW: {
            if (!stmt->data.drop_view.view_name || !txn->current_database) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            char* view_name = stmt->data.drop_view.view_name;
            bool if_exists = stmt->data.drop_view.if_exists;
            
            // Check if view exists
            bool view_exists = metadata_view_exists(txn->db->metadata, txn->current_database, view_name);
            
            if (!view_exists && !if_exists) {
                exec_result = GU_ERR_SYNTAX; // View doesn't exist and IF EXISTS not specified
                break;
            }
            
            if (view_exists) {
                // Remove view from metadata
                if (metadata_remove_view(txn->db->metadata, txn->current_database, view_name) == 0) {
                    // Log to WAL
                    if (txn->db->wal_manager) {
                        char log_msg[512];
                        snprintf(log_msg, sizeof(log_msg), "DROP VIEW %s", view_name);
                        
                        wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                     txn->xid, 1, 0, time(NULL),
                                     log_msg, strlen(log_msg));
                    }
                }
            }
            
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_CREATE_TRIGGER: {
            if (!stmt->data.create_trigger.trigger_name || 
                !stmt->data.create_trigger.table_name || 
                !stmt->data.create_trigger.action_sql || 
                !txn->current_database) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            // Check if trigger system is initialized
            if (!txn->db->trigger_system) {
                // Initialize trigger system with default config
                trigger_system_config_t config = trigger_system_default_config();
                txn->db->trigger_system = trigger_system_create(&config, txn->db);
                if (txn->db->trigger_system) {
                    trigger_system_init((trigger_system_t*)txn->db->trigger_system, NULL);
                }
            }
            
            if (!txn->db->trigger_system) {
                exec_result = GU_ERR_INTERNAL;
                break;
            }
            
            // Convert timing and events from parser format to trigger system format
            trigger_timing_t timing = (trigger_timing_t)stmt->data.create_trigger.timing;
            trigger_event_t events = (trigger_event_t)stmt->data.create_trigger.events;
            
            // Create trigger in trigger system
            int result = trigger_system_create_trigger(
                (trigger_system_t*)txn->db->trigger_system,
                stmt->data.create_trigger.trigger_name,
                stmt->data.create_trigger.table_name,
                events,
                timing,
                TRIGGER_ACTION_SQL,
                stmt->data.create_trigger.action_sql,
                stmt->data.create_trigger.condition
            );
            
            if (result == 0) {
                // Log to WAL
                if (txn->db->wal_manager) {
                    char log_msg[512];
                    snprintf(log_msg, sizeof(log_msg), 
                            "CREATE TRIGGER %s %s %s ON %s", 
                            stmt->data.create_trigger.trigger_name,
                            (timing == 0) ? "BEFORE" : (timing == 1) ? "AFTER" : "INSTEAD OF",
                            (events & 0x01) ? "INSERT" : (events & 0x02) ? "UPDATE" : "DELETE",
                            stmt->data.create_trigger.table_name);
                    
                    wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                 txn->xid, 1, 0, time(NULL),
                                 log_msg, strlen(log_msg));
                }
                exec_result = GU_OK;
            } else {
                exec_result = GU_ERR_SYNTAX; // Trigger creation failed
            }
            
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_DROP_TRIGGER: {
            if (!stmt->data.drop_trigger.trigger_name || !txn->current_database) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            // Check if trigger system exists
            if (!txn->db->trigger_system) {
                if (stmt->data.drop_trigger.if_exists) {
                    exec_result = GU_OK; // IF EXISTS specified, so ignore
                } else {
                    exec_result = GU_ERR_SYNTAX; // Trigger doesn't exist
                }
                break;
            }
            
            trigger_system_t* trigger_system = (trigger_system_t*)txn->db->trigger_system;
            
            // Check if trigger exists
            trigger_definition_t* trigger = trigger_system_find_trigger(
                trigger_system, stmt->data.drop_trigger.trigger_name);
            
            if (!trigger && !stmt->data.drop_trigger.if_exists) {
                exec_result = GU_ERR_SYNTAX; // Trigger doesn't exist and IF EXISTS not specified
                break;
            }
            
            if (trigger) {
                // Drop trigger from trigger system
                int result = trigger_system_drop_trigger(
                    trigger_system, stmt->data.drop_trigger.trigger_name);
                
                if (result == 0) {
                    // Log to WAL
                    if (txn->db->wal_manager) {
                        char log_msg[512];
                        snprintf(log_msg, sizeof(log_msg), "DROP TRIGGER %s", 
                                stmt->data.drop_trigger.trigger_name);
                        
                        wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                                     txn->xid, 1, 0, time(NULL),
                                     log_msg, strlen(log_msg));
                    }
                    exec_result = GU_OK;
                } else {
                    exec_result = GU_ERR_INTERNAL; // Drop failed
                }
            } else {
                exec_result = GU_OK; // IF EXISTS specified and trigger doesn't exist
            }
            
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_CREATE_USER: {
            if (!stmt->data.create_user.username) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            // Check if user already exists
            for (int i = 0; i < GUDB_MAX_USERS; i++) {
                if (strlen(txn->db->users[i].username) > 0 &&
                    strcmp(txn->db->users[i].username, stmt->data.create_user.username) == 0) {
                    exec_result = GU_ERR_CONFLICT; // User already exists
                    goto create_user_end;
                }
            }
            
            // Find empty slot for new user
            int user_slot = -1;
            for (int i = 0; i < GUDB_MAX_USERS; i++) {
                if (strlen(txn->db->users[i].username) == 0) {
                    user_slot = i;
                    break;
                }
            }
            
            if (user_slot == -1) {
                exec_result = GU_ERR_NOMEM; // No more user slots
                break;
            }
            
            // Create new user
            strncpy(txn->db->users[user_slot].username, 
                   stmt->data.create_user.username, 
                   sizeof(txn->db->users[user_slot].username) - 1);
            
            if (stmt->data.create_user.password) {
                // Hash the password (simplified - in real implementation would use proper crypto)
                snprintf(txn->db->users[user_slot].password_hash, 
                        sizeof(txn->db->users[user_slot].password_hash),
                        "HASH_%s", stmt->data.create_user.password);
            }
            
            if (stmt->data.create_user.role) {
                strncpy(txn->db->users[user_slot].role,
                       stmt->data.create_user.role,
                       sizeof(txn->db->users[user_slot].role) - 1);
            } else {
                strcpy(txn->db->users[user_slot].role, "READ_WRITE");
            }
            
            // Log to WAL
            if (txn->db->wal_manager) {
                char log_msg[512];
                snprintf(log_msg, sizeof(log_msg), "CREATE USER %s ROLE %s", 
                        stmt->data.create_user.username,
                        txn->db->users[user_slot].role);
                
                wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                             txn->xid, 1, 0, time(NULL),
                             log_msg, strlen(log_msg));
            }
            
            exec_result = GU_OK;
            
            create_user_end:
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        case STMT_GRANT: {
            if (!stmt->data.grant.username || !stmt->data.grant.table_name) {
                exec_result = GU_ERR_SYNTAX;
                break;
            }
            
            // Find the user
            int user_index = -1;
            for (int i = 0; i < GUDB_MAX_USERS; i++) {
                if (strlen(txn->db->users[i].username) > 0 &&
                    strcmp(txn->db->users[i].username, stmt->data.grant.username) == 0) {
                    user_index = i;
                    break;
                }
            }
            
            if (user_index == -1) {
                exec_result = GU_ERR_SYNTAX; // User not found
                break;
            }
            
            // Find existing permission entry or create new one
            int perm_slot = -1;
            for (int i = 0; i < GUDB_MAX_USERS * 10; i++) {
                if (strlen(txn->db->permissions[i].table) > 0 &&
                    strcmp(txn->db->permissions[i].table, stmt->data.grant.table_name) == 0) {
                    // Found existing permission for this table, check if it's for the same user
                    // (simplified - would need better user-permission association)
                    perm_slot = i;
                    break;
                }
                if (strlen(txn->db->permissions[i].table) == 0 && perm_slot == -1) {
                    perm_slot = i; // Remember first empty slot
                }
            }
            
            if (perm_slot == -1) {
                exec_result = GU_ERR_NOMEM; // No more permission slots
                break;
            }
            
            // Set up permissions
            strncpy(txn->db->permissions[perm_slot].table, 
                   stmt->data.grant.table_name,
                   sizeof(txn->db->permissions[perm_slot].table) - 1);
            
            uint8_t perms = stmt->data.grant.permissions;
            txn->db->permissions[perm_slot].select = (perms & GU_PERM_SELECT) != 0;
            txn->db->permissions[perm_slot].insert = (perms & GU_PERM_INSERT) != 0;
            txn->db->permissions[perm_slot].update = (perms & GU_PERM_UPDATE) != 0;
            txn->db->permissions[perm_slot].delete = (perms & GU_PERM_DELETE) != 0;
            
            // Log to WAL
            if (txn->db->wal_manager) {
                char log_msg[512];
                char perm_str[100] = "";
                if (perms & GU_PERM_SELECT) strcat(perm_str, "SELECT ");
                if (perms & GU_PERM_INSERT) strcat(perm_str, "INSERT ");
                if (perms & GU_PERM_UPDATE) strcat(perm_str, "UPDATE ");
                if (perms & GU_PERM_DELETE) strcat(perm_str, "DELETE ");
                
                snprintf(log_msg, sizeof(log_msg), "GRANT %s ON %s TO %s", 
                        perm_str, stmt->data.grant.table_name, stmt->data.grant.username);
                
                wal_log_insert((wal_manager_t*)txn->db->wal_manager, 
                             txn->xid, 1, 0, time(NULL),
                             log_msg, strlen(log_msg));
            }
            
            exec_result = GU_OK;
            res->row_count = 0;
            res->col_count = 0;
            break;
        }
        
        default:
            exec_result = GU_ERR_SYNTAX;
            break;
    }
    
    // Cleanup
    statement_destroy(stmt);
    parser_destroy(parser);
    
    if (exec_result != GU_OK) {
        gudb_free_result(res);
        return exec_result;
    }
    
    *result = res;
    return GU_OK;
}

// Free result
void gudb_free_result(result_t* result) {
    if (!result) return;

    if (result->column_names) {
        for (uint32_t i = 0; i < result->col_count; i++) {
            free(result->column_names[i]);
        }
        free(result->column_names);
    }

    if (result->rows) {
        for (uint32_t i = 0; i < result->row_count; i++) {
            if (result->rows[i]) {
                for (uint32_t j = 0; j < result->col_count; j++) {
                    free(result->rows[i][j]);
                }
                free(result->rows[i]);
            }
        }
        free(result->rows);
    }

    if (result->error_msg) {
        free(result->error_msg);
    }

    free(result);
}

// Error functions
const char* gudb_error(gudb* db __attribute__((unused))) {
    return last_error[0] ? last_error : "No error";
}

const char* gudb_txn_error(gudb_txn* txn __attribute__((unused))) {
    return last_error[0] ? last_error : "No error";
}

const char* gudb_txn_current_database(gudb_txn* txn) {
    if (!txn) return NULL;
    return txn->current_database;
}

// Internal helper functions implementation


static int init_page_cache(gudb* db) {
    db->page_cache = malloc(GUDB_PAGE_CACHE_SIZE);
    if (!db->page_cache) {
        set_error("Failed to allocate page cache");
        return GU_ERR_NOMEM;
    }
    memset(db->page_cache, 0, GUDB_PAGE_CACHE_SIZE);
    return GU_OK;
}

static int init_wal_system(gudb* db) {
    db->wal_buffer = malloc(GUDB_WAL_BUFFER_SIZE);
    if (!db->wal_buffer) {
        set_error("Failed to allocate WAL buffer");
        return GU_ERR_NOMEM;
    }
    
    db->binlog_buffer = malloc(GUDB_WAL_BUFFER_SIZE);
    if (!db->binlog_buffer) {
        set_error("Failed to allocate binlog buffer");
        return GU_ERR_NOMEM;
    }
    
    memset(db->wal_buffer, 0, GUDB_WAL_BUFFER_SIZE);
    memset(db->binlog_buffer, 0, GUDB_WAL_BUFFER_SIZE);
    
    return GU_OK;
}

static int load_metadata(gudb* db) {
    // Read metadata file to load users and permissions
    // For now, create a default admin user
    strcpy(db->users[0].username, "admin");
    strcpy(db->users[0].password_hash, "admin_hash");  // Should be proper hash
    strcpy(db->users[0].role, "admin");
    
    return GU_OK;
}

static int authenticate_user(gudb* db, const char* username, const char* password __attribute__((unused))) {
    // Simple authentication for now
    for (int i = 0; i < GUDB_MAX_USERS && db->users[i].username[0]; i++) {
        if (strcmp(db->users[i].username, username) == 0) {
            // In real implementation, hash password and compare
            return GU_OK;
        }
    }
    
    set_error("Authentication failed");
    return GU_ERR_AUTH;
}

static void set_error(const char* msg) {
    strncpy(last_error, msg, sizeof(last_error) - 1);
    last_error[sizeof(last_error) - 1] = '\0';
}

// CRC32 implementation (for future use)
// static uint32_t crc32(const void* data, size_t len) {
//     const uint8_t* bytes = (const uint8_t*)data;
//     uint32_t crc = 0xFFFFFFFF;
//     
//     for (size_t i = 0; i < len; i++) {
//         crc ^= bytes[i];
//         for (int j = 0; j < 8; j++) {
//             if (crc & 1) {
//                 crc = (crc >> 1) ^ 0xEDB88320;
//             } else {
//                 crc >>= 1;
//             }
//         }
//     }
//     
//     return crc ^ 0xFFFFFFFF;
// }

static uint64_t get_timestamp_ms(void) {
    struct timespec ts;
    clock_gettime(CLOCK_REALTIME, &ts);
    return (uint64_t)ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
}

// User management implementation
int gudb_create_user(gudb* db, const char* username, const char* password, const char* role) {
    if (!db || !username || !password || !role) {
        return GU_ERR_SYNTAX;
    }
    
    // Check if user already exists
    for (int i = 0; i < GUDB_MAX_USERS; i++) {
        if (strlen(db->users[i].username) > 0 && strcmp(db->users[i].username, username) == 0) {
            return GU_ERR_SYNTAX; // User already exists
        }
    }
    
    // Find empty slot for new user
    int user_slot = -1;
    for (int i = 0; i < GUDB_MAX_USERS; i++) {
        if (strlen(db->users[i].username) == 0) {
            user_slot = i;
            break;
        }
    }
    
    if (user_slot == -1) {
        return GU_ERR_NOMEM; // No more user slots available
    }
    
    // Validate role
    if (strcmp(role, "ADMIN") != 0 && strcmp(role, "READ_WRITE") != 0 && strcmp(role, "READ_ONLY") != 0) {
        return GU_ERR_SYNTAX; // Invalid role
    }
    
    // Create new user
    strncpy(db->users[user_slot].username, username, sizeof(db->users[user_slot].username) - 1);
    strncpy(db->users[user_slot].role, role, sizeof(db->users[user_slot].role) - 1);
    
    // Simple password hashing (in production, use proper hashing like bcrypt)
    // For now, we'll just use a simple hash
    char password_hash[64];
    snprintf(password_hash, sizeof(password_hash), "hash_%s_%lu", password, strlen(password));
    strncpy(db->users[user_slot].password_hash, password_hash, sizeof(db->users[user_slot].password_hash) - 1);
    
    return GU_OK;
}

int gudb_grant_permission(gudb* db, const char* username, const char* table, 
                         bool select, bool insert, bool update, bool delete) {
    if (!db || !username || !table) {
        return GU_ERR_SYNTAX;
    }
    
    // Find the user
    int user_index = -1;
    for (int i = 0; i < GUDB_MAX_USERS; i++) {
        if (strlen(db->users[i].username) > 0 && strcmp(db->users[i].username, username) == 0) {
            user_index = i;
            break;
        }
    }
    
    if (user_index == -1) {
        return GU_ERR_SYNTAX; // User not found
    }
    
    // Find existing permission or empty slot
    int perm_slot = -1;
    for (int i = 0; i < GUDB_MAX_USERS * 10; i++) {
        if (strlen(db->permissions[i].table) > 0 && strcmp(db->permissions[i].table, table) == 0) {
            // Check if this permission belongs to the user by checking position
            // Simple mapping: user_index * 10 + table_slot
            if (i >= user_index * 10 && i < (user_index + 1) * 10) {
                perm_slot = i;
                break;
            }
        }
    }
    
    // If no existing permission found, find empty slot for this user
    if (perm_slot == -1) {
        for (int i = user_index * 10; i < (user_index + 1) * 10; i++) {
            if (strlen(db->permissions[i].table) == 0) {
                perm_slot = i;
                break;
            }
        }
    }
    
    if (perm_slot == -1) {
        return GU_ERR_NOMEM; // No more permission slots available for this user
    }
    
    // Set up permissions
    strncpy(db->permissions[perm_slot].table, table, sizeof(db->permissions[perm_slot].table) - 1);
    db->permissions[perm_slot].select = select;
    db->permissions[perm_slot].insert = insert;
    db->permissions[perm_slot].update = update;
    db->permissions[perm_slot].delete = delete;
    
    return GU_OK;
}

int gudb_analyze_table(gudb* db, const char* table) {
    if (!db || !table) {
        return GU_ERR_SYNTAX;
    }
    
    // Start a temporary transaction for analysis
    gudb_txn* analysis_txn = gudb_begin(db, GU_TXN_RC, "");
    if (!analysis_txn) {
        return GU_ERR_NOMEM;
    }
    
    // Check if table exists in metadata
    if (!metadata_table_exists(db->metadata, analysis_txn->current_database, table)) {
        gudb_rollback(analysis_txn);
        return GU_ERR_SYNTAX; // Table not found
    }
    
    // Get table column information
    column_info_t* columns = NULL;
    uint32_t column_count = 0;
    if (metadata_get_table_columns(db->metadata, analysis_txn->current_database, table, &columns, &column_count) != 0 || !columns) {
        gudb_rollback(analysis_txn);
        return GU_ERR_IO;
    }
    
    // Analyze table data by scanning storage
    uint64_t row_count = 0;
    uint64_t total_size = 0;
    uint32_t max_rows_to_scan = 10000; // Limit for performance
    
    for (uint32_t row_id = 1; row_id <= max_rows_to_scan; row_id++) {
        char search_key[256];
        snprintf(search_key, sizeof(search_key), "%s:%u", table, row_id);
        
        search_result_t search_result = storage_search((bw_tree_t*)db->storage, 
                                                    search_key, strlen(search_key));
        if (search_result.found && search_result.value) {
            row_count++;
            total_size += strlen(search_result.value);
        }
    }
    
    // Update table statistics in metadata
    if (metadata_update_row_count(db->metadata, analysis_txn->current_database, table, row_count) == 0) {
        // Statistics updated successfully
    }
    
    // Log analysis results
    printf("ANALYZE TABLE %s completed:\n", table);
    printf("  Rows: %llu\n", (unsigned long long)row_count);
    printf("  Columns: %u\n", column_count);
    printf("  Estimated size: %llu bytes\n", (unsigned long long)total_size);
    printf("  Average row size: %.2f bytes\n", 
           row_count > 0 ? (double)total_size / row_count : 0.0);
    
    // Clean up
    free(columns);
    gudb_commit(analysis_txn);
    
    return GU_OK;
}

int gudb_checkpoint(gudb* db) {
    if (!db) {
        return GU_ERR_SYNTAX;
    }
    
    printf("Starting checkpoint operation...\n");
    
    // 1. Flush WAL buffer to disk
    if (db->wal_manager) {
        printf("Flushing WAL buffer to disk...\n");
        // Force WAL flush - assuming WAL manager has flush function
        // In a full implementation, this would call wal_flush()
    }
    
    // 2. Flush page cache to storage files
    if (db->page_cache) {
        printf("Flushing page cache to storage...\n");
        // Force dirty pages to disk - assuming page cache has flush function
        // In a full implementation, this would call page_cache_flush()
    }
    
    // 3. Sync metadata to disk
    if (db->metadata) {
        printf("Syncing metadata to disk...\n");
        // Force metadata flush - the metadata manager should have persistence
        // In a full implementation, this would call metadata_sync()
    }
    
    // 4. Sync file system changes
    if (db->fs) {
        printf("Syncing file system changes...\n");
        // Force all file system changes to disk
        // Call fsync on all open file descriptors
        if (db->data_fd > 0) {
            fsync(db->data_fd);
        }
        if (db->wal_fd > 0) {
            fsync(db->wal_fd);
        }
        if (db->meta_fd > 0) {
            fsync(db->meta_fd);
        }
    }
    
    // 5. Update checkpoint metadata
    uint64_t checkpoint_time = get_timestamp_ms();
    printf("Checkpoint completed at timestamp: %llu\n", (unsigned long long)checkpoint_time);
    
    // Log checkpoint completion
    if (db->wal_manager) {
        // Log the checkpoint event to WAL
        char checkpoint_msg[256];
        snprintf(checkpoint_msg, sizeof(checkpoint_msg), "CHECKPOINT completed at %llu", (unsigned long long)checkpoint_time);
        // In a full implementation, this would call wal_log_checkpoint()
    }
    
    printf("Checkpoint operation completed successfully.\n");
    return GU_OK;
}

int gudb_vacuum(gudb* db) {
    if (!db) {
        return GU_ERR_SYNTAX;
    }
    
    printf("Starting VACUUM operation...\n");
    
    // 1. Reclaim unused storage space
    uint64_t total_space_reclaimed = 0;
    uint64_t total_records_processed = 0;
    
    // Start a temporary transaction for vacuum operations
    gudb_txn* vacuum_txn = gudb_begin(db, GU_TXN_RC, "");
    if (!vacuum_txn) {
        return GU_ERR_NOMEM;
    }
    
    // 2. Get all tables in current database for processing
    table_info_t* tables = NULL;
    uint32_t table_count = 0;
    if (metadata_list_tables(db->metadata, vacuum_txn->current_database, &tables, &table_count) == 0 && tables) {
        
        printf("Processing %u tables for VACUUM...\n", table_count);
        
        for (uint32_t i = 0; i < table_count; i++) {
            const char* table_name = tables[i].table_name;
            printf("  Vacuuming table: %s\n", table_name);
            
            // Get table row count for optimization
            uint64_t estimated_rows = metadata_get_table_row_count(db->metadata, 
                                                                  vacuum_txn->current_database, 
                                                                  table_name);
            
            // 3. Scan and compact table data
            uint32_t valid_records = 0;
            uint32_t deleted_records = 0;
            uint32_t max_scan_limit = (estimated_rows > 0) ? estimated_rows + 1000 : 10000;
            
            for (uint32_t row_id = 1; row_id <= max_scan_limit; row_id++) {
                char search_key[256];
                snprintf(search_key, sizeof(search_key), "%s:%u", table_name, row_id);
                
                search_result_t search_result = storage_search((bw_tree_t*)db->storage, 
                                                            search_key, strlen(search_key));
                
                if (search_result.found && search_result.value) {
                    valid_records++;
                    total_records_processed++;
                } else {
                    // This represents a gap or deleted record
                    deleted_records++;
                }
                
                // Stop scanning if we've gone far beyond the estimated row count
                if (row_id > estimated_rows && deleted_records > valid_records) {
                    break;
                }
            }
            
            // Update table statistics after vacuum
            if (valid_records > 0) {
                metadata_update_row_count(db->metadata, vacuum_txn->current_database, 
                                               table_name, valid_records);
            }
            
            printf("    Valid records: %u, Deleted gaps: %u\n", valid_records, deleted_records);
            total_space_reclaimed += deleted_records * 100; // Estimated 100 bytes per deleted record
        }
        
        free(tables);
    }
    
    // 4. Optimize storage structure
    if (db->storage) {
        printf("Optimizing storage structure...\n");
        // In a full implementation, this would call storage optimization functions
        // such as bw_tree_compact() or similar storage reorganization
    }
    
    // 5. Flush changes and update metadata
    if (db->metadata) {
        printf("Updating metadata statistics...\n");
        metadata_save_to_disk(db->metadata);
    }
    
    // 6. Force checkpoint after vacuum
    printf("Performing checkpoint after VACUUM...\n");
    int checkpoint_result = gudb_checkpoint(db);
    
    // 7. Report vacuum results
    printf("VACUUM operation completed:\n");
    printf("  Total records processed: %llu\n", (unsigned long long)total_records_processed);
    printf("  Estimated space reclaimed: %llu bytes\n", (unsigned long long)total_space_reclaimed);
    printf("  Checkpoint result: %s\n", (checkpoint_result == GU_OK) ? "SUCCESS" : "FAILED");
    
    // Clean up transaction
    gudb_commit(vacuum_txn);
    
    return GU_OK;
}

int gudb_server_start(gudb* db, const char* host __attribute__((unused)), uint16_t port) {
    if (!db) return GU_ERR_SYNTAX;
    
    // Don't start if already running
    if (db->network_server) {
        return GU_OK;
    }
    
    // Create network server
    db->network_server = network_server_create(port, db);
    if (!db->network_server) {
        return GU_ERR_NOMEM;
    }
    
    // Start network server
    if (network_server_start((network_server_t*)db->network_server) != 0) {
        network_server_destroy((network_server_t*)db->network_server);
        db->network_server = NULL;
        return GU_ERR_IO;
    }
    
    printf("GuDB server started on port %u\n", port);
    return GU_OK;
}

int gudb_server_stop(gudb* db) {
    if (!db || !db->network_server) return GU_OK;
    
    // Stop network server
    network_server_stop((network_server_t*)db->network_server);
    network_server_destroy((network_server_t*)db->network_server);
    db->network_server = NULL;
    
    printf("GuDB server stopped\n");
    return GU_OK;
}

// Helper function to create tuple from JSON data for WHERE clause evaluation
static tuple_t* create_tuple_from_json(const char* json_data, column_info_t* columns, uint32_t column_count) {
    if (!json_data || !columns || column_count == 0) {
        return NULL;
    }
    
    // Create tuple
    tuple_t* tuple = malloc(sizeof(tuple_t));
    if (!tuple) {
        return NULL;
    }
    
    // Create tuple slot
    tuple->slot = malloc(sizeof(tuple_slot_t));
    if (!tuple->slot) {
        free(tuple);
        return NULL;
    }
    
    tuple->slot->column_count = column_count;
    tuple->slot->values = malloc(sizeof(value_t) * column_count);
    if (!tuple->slot->values) {
        free(tuple->slot);
        free(tuple);
        return NULL;
    }
    
    // Parse JSON and extract column values
    for (uint32_t i = 0; i < column_count; i++) {
        char* column_name = columns[i].column_name;
        
        // Find column value in JSON string
        char pattern[256];
        snprintf(pattern, sizeof(pattern), "\"%s\":\"", column_name);
        char* start = strstr(json_data, pattern);
        
        if (start) {
            start += strlen(pattern);
            char* end = strchr(start, '"');
            if (end) {
                // Extract value
                size_t value_len = end - start;
                char value_str[256];
                strncpy(value_str, start, value_len);
                value_str[value_len] = '\0';
                
                // Convert to appropriate data type based on column type
                switch (columns[i].data_type) {
                    case DATA_TYPE_INT:
                    case DATA_TYPE_BIGINT:
                        tuple->slot->values[i] = value_make_int64(atoll(value_str));
                        break;
                    case DATA_TYPE_FLOAT:
                    case DATA_TYPE_DOUBLE:
                        tuple->slot->values[i] = value_make_double(atof(value_str));
                        break;
                    case DATA_TYPE_VARCHAR:
                    case DATA_TYPE_CHAR:
                    case DATA_TYPE_TEXT:
                    default:
                        tuple->slot->values[i] = value_make_string(value_str);
                        break;
                }
            } else {
                // Default null value
                tuple->slot->values[i] = value_make_null();
            }
        } else {
            // Column not found in JSON, set to null
            tuple->slot->values[i] = value_make_null();
        }
    }
    
    return tuple;
}